package com.ruoyi.business.interceptor;

import com.ruoyi.business.config.MqttConfig;
import com.ruoyi.business.config.PushCallback;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;

@Component
public class MqttPushClient {
    private static final Logger logger = LoggerFactory.getLogger(MqttPushClient.class);
    private static MqttClient client;

    @Autowired
    private PushCallback pushCallback;

    @Autowired
    private MqttConfig mqttConfig;

    public static MqttClient getClient() {
        return client;
    }

    private static void setClient(MqttClient client) {
        MqttPushClient.client = client;
    }

    /**
     * 客户端连接
     */
    public void connect() {
        MqttClient client;
        try {
            client = new MqttClient(mqttConfig.getHostUrl(), mqttConfig.getInClientId(), new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            //接受离线消息。true：每次连接服务器都是以新的身份 false：告诉代理客户端是否要建立持久会话   false为建立持久会话
            options.setCleanSession(mqttConfig.getClearSession());
            //设置连接用户名
            options.setUserName(mqttConfig.getUsername());
            //设置连接密码
            options.setPassword(mqttConfig.getPassword().toCharArray());
            //设置超时时间，单位为秒
            options.setConnectionTimeout(mqttConfig.getTimeout());
            //设置心跳时间 单位为秒，表示服务器每隔 1.5*20秒的时间向客户端发送心跳判断客户端是否在线
            options.setKeepAliveInterval(mqttConfig.getKeepalive());
            //设置自动链接
            options.setAutomaticReconnect(mqttConfig.getAutoReconnect());
            //设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息
            mqttConfig.setWillTopic(mqttConfig.getServerStateTopic());
            options.setMaxInflight(1000);
            options.setWill(mqttConfig.getWillTopic(), (mqttConfig.getInClientId() + "与服务器断开连接").getBytes(), mqttConfig.getWillQos(), mqttConfig.getWillRetained());
            MqttPushClient.setClient(client);
            try {
                //设置回调
                client.setCallback(pushCallback);
                if (!client.isConnected()) {
                    client.connect(options);
                    logger.info(mqttConfig.getInClientId() + "尝试建立新连接");
                } else {
                    client.disconnect();
                    client.connect(options);
                    logger.info(mqttConfig.getInClientId() + "尝试重新连接");
                    logger.info("===>>>MQTT断开尝试重连成功<<<===");
                }
            } catch (Exception e) {
                logger.error("MQTT连接失败:" + e.getMessage());
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 发布
     *
     * @param retained    是否保留
     * @param topic       主题
     * @param pushMessage 消息体
     */
    public void publish(boolean retained, String topic, String pushMessage) throws MqttException {
        MqttMessage message = new MqttMessage();
        //连接方式,为0非持久化
        message.setQos(0);
        //是否保留
        message.setRetained(retained);

        logger.info("Send to Mqtt -- topic : {}, message : {}", topic, pushMessage);
//        message.setPayload(pushMessage.getBytes(StandardCharsets.UTF_8));
        try {
            message.setPayload(pushMessage.getBytes("UTF-8"));
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        MqttTopic mTopic = getClient().getTopic(topic);
        if (null == mTopic) {
            logger.error("主题不存在:{}", topic);
        }
        mTopic.publish(message);
    }

    /**
     * 订阅某个主题
     *
     * @param topic 主题
     * @param qos   连接方式
     */
    public void subscribe(String topic, int qos) {
        logger.info("开始订阅主题" + topic);
        try {
            client.unsubscribe(topic);
            client.subscribe(topic, qos);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 取消订阅主题
     *
     * @param topic 主题名称
     */
    public void cleanTopic(String topic) {
        if (client != null && client.isConnected()) {
            try {
                client.unsubscribe(topic);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        } else {
            System.out.println("取消订阅失败！");
        }
    }
}
